package com.king.util

import java.util.Properties

import org.apache.flink.api.common.serialization.{DeserializationSchema, SimpleStringSchema}
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.KafkaProducer

/**
 * @Author: KingWang
 * @Date: 2022/1/24  
 * @Desc:
 **/
object MyKafkaUtil {

  val brokerList = "hadoop200:9092,hadoop201:9092,hadoop202:9092"

  def getKafkaProducer(topic:String):FlinkKafkaProducer[String] = {
    new FlinkKafkaProducer[String](brokerList:String,topic:String, new SimpleStringSchema())
  }

  def getKafkaConsumer(topic:String,groupId:String):FlinkKafkaConsumer[String] = {
    val prop = new Properties()
    prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,brokerList)
    prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId)
    new FlinkKafkaConsumer[String](topic,new SimpleStringSchema(), prop)
  }

}
